defmodule SockWorkstation.SockVOASRClient do
  # use GenServer
  use WebSockex
  require Logger

  @default_interval 0
  @protocol_version 0b0001
  @default_header_size 0b0001

  # Message Types
  @full_client_request 0b0001
  @audio_only_request 0b0010
  @full_server_response 0b1001
  @server_ack 0b1011
  @server_error_response 0b1111

  # Flags
  # @audio_res 0b0000
  @audio_sequence_rsp 0b0001
  # @last_res 0b0010
  @last_sequence_rep 0b0011

  # Serialization
  # @no_serialization 0b0000
  @json 0b0001

  # Compression
  # @no_compression 0b0000
  @gzip 0b0001

  def start_link(session_id) do
    config = %{
      url: "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_nostream",
      app_id: "5712898771",
      token: "DM1O_wgeItdJocj1Zx7A4nVdM-v_vBOv"
    }

    WebSockex.start_link(config.url, __MODULE__, %{session_id: session_id},
      extra_headers: headers(config)
      # debug: [:trace]
    )
  end

  @impl true
  def handle_connect(_conn, state) do
    self_pid = self()

    Logger.info("Connected to server")
    Process.send_after(self_pid, :full_client_request, 240)
    run_append_pcmdata(self_pid)

    state = %{
      session_id: state.session_id,
      frames: [],
      receiver_pid: nil,
      self_pid: self_pid,
      interval: @default_interval,
      timer_ref: nil,
      seq: 1,
      sample_rate: 16000,
      bits: 16,
      channels: 1,
      last_sent: false
    }

    # state
    # |> Map.put(:sample_rate, 16000)
    # |> Map.put(:bits, 16)
    # |> Map.put(:channels, 1)
    # |> Map.put(:seq, 1)
    # |> Map.put(:buffer, <<>>)
    # # |> Map.put(:audio_stream, File.stream!("votts.pcm", [], 1920))
    # |> Map.put(:last_sent, false)
    # |> Map.put(:self_pid, self_pid)

    # send_full_client_request(state)
    {:ok, state}
  end

  @impl true
  def handle_frame({:binary, data}, state) do
    Logger.info("Received #{byte_size(data)} bytes")

    case parse_response(data) do
      {:ok, result} ->
        Logger.info("Received response: #{inspect(result)}")
        Process.sleep(200)
        # send_audio_request(state)
        # run_audio_client_request(state.self_pid)

        # Process.sleep(1000)
        send(state.self_pid, :start_push)
        {:ok, %{state | seq: result.sequence}}

      {:error, reason} ->
        Logger.error("Error processing response: #{reason}")
        {:close, state}

        # :done ->
        #   Logger.info("ASR processing completed")
        #   {:close, state}
    end
  end

  @impl true
  def handle_disconnect(%{reason: reason}, state) do
    Logger.info("Disconnected: #{inspect(reason)}")
    {:ok, state}
  end

  defp headers(config) do
    [
      {"X-Api-App-Key", config.app_id},
      {"X-Api-Access-Key", config.token},
      {"X-Api-Resource-Id", "volc.bigasr.sauc.duration"},
      {"X-Api-Connect-Id", UUID.uuid4()}
    ]
  end

  # def full_client_request(pid) do
  #   send(pid, :full_client_request)
  # end

  def run_append_pcmdata(pid) do
    audio_stream = File.stream!("tts.pcm", [], 7200)

    Enum.each(audio_stream, fn pcm_data ->
      #  Process.sleep(100)
      send(pid, {:append_frame, pcm_data})
    end)

    # send(pid, :last_client_request)
  end

  @impl true
  def handle_info(:full_client_request, state) do
    payload =
      %{
        user: %{uid: "test"},
        audio: %{
          format: "pcm",
          sample_rate: state.sample_rate,
          bits: state.bits,
          channel: state.channels,
          language: "zh-CN"
          # codec: "raw"
        },
        request: %{
          model_name: "bigmodel",
          enable_punc: true
        }
      }
      |> Jason.encode!()

    compressed = :zlib.gzip(payload)
    header = build_header(@full_client_request, @audio_sequence_rsp, @json, @gzip)
    seq_bytes = <<state.seq::32>>
    payload_size = <<byte_size(compressed)::32>>

    frame =
      [
        header,
        seq_bytes,
        payload_size,
        compressed
      ]
      |> IO.iodata_to_binary()

    #  IO.inspect(frame)

    {:reply, {:binary, frame}, state}
  end

  def handle_info(:audio_client_request, %{frames: [frame | rest]} = state) do
    # audio_stream = File.stream!("tts.pcm")

    # case Enum.take(state.audio_stream, 1) do
    # pcm_data when is_binary(pcm_data) ->
    # {seq, flags} = next_sequence(state)
    sequence = state.seq + 1
    Logger.info("Sending audio frame with sequence: #{sequence}")
    compressed = :zlib.gzip(frame)
    header = build_header(@audio_only_request, @audio_sequence_rsp, @json, @gzip)
    seq_bytes = <<sequence::32>>
    payload_size = <<byte_size(compressed)::32>>

    frame =
      [
        header,
        seq_bytes,
        payload_size,
        compressed
      ]
      |> IO.iodata_to_binary()

    new_state = %{
      state
      | seq: sequence,
        frames: rest
        # last_sent: seq < 0,
        # buffer: <<>>
        # audio_stream: Stream.drop(state.audio_stream, 1)
    }

    # Process.sleep(200)
    {:reply, {:binary, frame}, schedule_next_frame(new_state)}

    # _ ->
    # Logger.info("Audio stream ended")
    # {:close, state}
    # end
  end

  def handle_info(:last_client_request, state) do
    # audio_stream = File.stream!("tts.pcm")

    # case Enum.take(state.audio_stream, 1) do
    # pcm_data when is_binary(pcm_data) ->
    # {seq, flags} = next_sequence(state)
    # Logger.info("Sending audio frame with sequence: #{seq}")
    pcm_data = :binary.copy(<<0>>, 1920)
    compressed = :zlib.gzip(pcm_data)

    header = build_header(@audio_only_request, @last_sequence_rep, @json, @gzip)
    seq_bytes = <<-(state.seq + 1)::32>>
    payload_size = <<byte_size(compressed)::32>>

    frame =
      [
        header,
        seq_bytes,
        payload_size,
        compressed
      ]
      |> IO.iodata_to_binary()

    new_state = %{
      state
      | # | seq: abs(seq),
        last_sent: true
        # buffer: <<>>
        # audio_stream: Stream.drop(state.audio_stream, 1)
    }

    # Process.sleep(200)
    {:reply, {:binary, frame}, new_state}

    # _ ->
    # Logger.info("Audio stream ended")
    # {:close, state}
    # end
  end

  #    def handle_info({:queue_frames, frames}, state) do
  #     new_state = %{state | frames: state.frames ++ frames}
  #     {:noreply, new_state}
  #  end

  def handle_info(:start_push, state) do
    case state.receiver_pid do
      nil ->
        new_state = schedule_next_frame(%{state | receiver_pid: state.self_pid})
        {:ok, new_state}

      _ ->
        {:ok, state}
    end
  end

  # def handle_info(:stop_push, state) do
  #   send(state.self_pid, :last_client_request)
  #   state =
  #     state
  #     |> cancel_timer()
  #     |> Map.put(:receiver_pid, nil)

  #   {:ok, state}
  # end

  def handle_info({:append_frame, frame}, state) do
    # new_state = %{state | frames: state.frames ++ [frame]}
    # schedule_next_frame(state)
    new_state = %{state | frames: [frame | Enum.reverse(state.frames)] |> Enum.reverse()}
    {:ok, new_state}
  end

  def handle_info(:push_next, %{frames: []} = state) do
    Logger.info("No frames to push")
    # send(state.self_pid, :stop_push)

    send(state.self_pid, :last_client_request)

    state =
      state
      |> cancel_timer()
      |> Map.put(:receiver_pid, nil)

    {:ok, state}
  end

  def handle_info(:push_next, state) do
    send(state.self_pid, :audio_client_request)
    {:ok, state}
  end

  # defp next_sequence(%{last_sent: true} = _state) do
  #   # 已经发送过结束标记，不再发送新数据
  #   {:halt, nil}
  # end

  # defp next_sequence(state) do
  #   # 添加流结束检测函数
  #   # is_last = is_stream_ended?(state)
  #   # seq = state.seq + 1

  #   if state.last_sent do
  #     {-state.seq, @last_res}
  #   else
  #     {state.seq + 1, @audio_res}
  #   end
  # end

  defp build_header(message_type, flags, serialization, compression) do
    # header =
    <<
      @protocol_version::4,
      @default_header_size::4,
      message_type::4,
      flags::4,
      serialization::4,
      compression::4,
      # Reserved
      0::8
    >>

    #  Logger.debug("""
    # Built header:
    # - Message type: #{message_type}
    # - Flags: #{flags}
    # - Serialization: #{serialization}
    # - Compression: #{compression}
    # Raw: #{inspect(header)}
    # """)
    # header
  end

  defp parse_response(<<
         _protocol::4,
         _header_size::4,
         message_type::4,
         _flags::4,
         _serialization::4,
         compression::4,
         _reserved::8,
         seq::32,
         _payload_size::32,
         payload::binary
       >>) do
    case message_type do
      @full_server_response ->
        process_payload(payload, compression, seq)

      @server_ack ->
        Logger.info("Server ACK received")
        {:ok, %{seq: seq}}

      @server_error_response ->
        error_msg = if compression == @gzip, do: :zlib.gunzip(payload), else: payload
        {:error, "Server error #{seq}: #{error_msg}"}

      _ ->
        {:error, "Unknown message type #{message_type}"}
    end
  end

  defp process_payload(payload, @gzip, seq) do
    case :zlib.gunzip(payload) do
      decompressed when is_binary(decompressed) ->
        case Jason.decode(decompressed) do
          {:ok, json} -> {:ok, %{sequence: seq, data: json}}
          {:error, _} -> {:ok, %{sequence: seq, data: decompressed}}
        end

      _ ->
        {:error, "Failed to decompress payload"}
    end
  end

  defp process_payload(payload, _, seq) do
    {:ok, %{sequence: seq, data: payload}}
  end

  defp schedule_next_frame(state) do
    timer_ref = Process.send_after(self(), :push_next, state.interval)
    %{state | timer_ref: timer_ref}
  end

  #  defp send_frame(%{frames: [frame | rest]} = state) do
  #     case SockCoopClient.send_binary(state.coop_pid, frame) do
  #       :ok ->

  #         {:ok, %{state | frames: rest}}

  #       {:error, reason} ->
  #         {:error, reason, state}
  #     end
  #   end

  defp cancel_timer(%{timer_ref: nil} = state), do: state

  defp cancel_timer(state) do
    Process.cancel_timer(state.timer_ref)
    %{state | timer_ref: nil}
  end
end

# defp parse_response(<<
#        _protocol::4,
#        _header_size::4,
#        message_type::4,
#        _flags::4,
#        _serialization::4,
#        compression::4,
#        _reserved::8,
#        seq::32,
#        payload_size::32,
#        payload::binary
#      >>) do
#   case message_type do
#     @full_server_response ->
#       decompress_payload(payload, compression)
#       seq

#     @server_ack ->
#       decompress_payload(payload, compression)
#       seq

#     @server_error_response ->
#       {:error, seq, payload}

#     _ ->
#       seq
#   end
# end

# defp decompress_payload(payload, @gzip) do
#   payload
#   |> :zlib.gunzip()
#   |> Jason.decode!()
#   |> Logger.info()
# end

# defp decompress_payload(payload, _), do: payload

# def send_audio_request(pid, pcm_data) do
# case Enum.at(state.audio_stream, 1) do
# data when is_binary(data) ->
# {seq, flags} = next_sequence(state)
# compressed = :zlib.gzip(data)
# header = build_header(@audio_only_request, flags, @json, @gzip)
# seq_bytes = <<seq::32>>
# payload_size = <<byte_size(compressed)::32>>

# frame =
#   [
#     header,
#     seq_bytes,
#     payload_size,
#     compressed
#   ]
#   |> IO.iodata_to_binary()

# new_state = %{
#   state
#   | seq: abs(seq),
#     last_sent: seq < 0,
#     audio_stream: Stream.drop(state.audio_stream, 1)
# }
# send(state.self_pid, {:binary, frame})
# {:reply, {:binary, frame}, new_state}
# WebSockex.send_frame(pid, {:binary, frame})

# _ ->
# Logger.info("Audio stream ended")
# {:close, state}
# end
# end
